package com.ccf.business.etl.flow.mq;

import com.alibaba.fastjson.JSON;
import com.component.api.constants.FlowConstants;
import com.component.api.model.flow.FlowInfo;
import com.framework.rocketmq.producer.RocketMQProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

/**
 * 描述：
 * 流程运行消息队列发送者
 * @author xianggj
 * @Date 2021/11/30 14:08
 **/
@Slf4j
@Service
public class FlowRunningProducer {

    @Autowired
    private RocketMQProducer rocketMQProducer;


    public void running(FlowInfo flowInfo) {
        log.info("流程运行发送消息：");
        Message sendMsg = new Message(FlowConstants.FLOW_TOPIC, "*", JSON.toJSONBytes(flowInfo));
        try {
            sendMessage(sendMsg);
        } catch (RemotingException | MQClientException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * @param message 消息体
     * @apiNote mq发送消息
     * @author yepk
     * @date 2019/3/12 8:58
     */
    private void sendMessage(Message message) throws RemotingException, MQClientException, InterruptedException {


        //默认3秒超时
        rocketMQProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送完成");
                log.info(JSON.toJSONString(sendResult));
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送失败");
                throwable.printStackTrace();
            }
        });
    }
}
